package mya.mes.mdc.websocket;

import com.fasterxml.jackson.databind.node.ArrayNode;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;

//import util.HTMLFilter;

/**
 * 服务端广播：后台管理界面等
 */
@ServerEndpoint(value = "/websocket/mdc/")
public class MdcBroadcast {

    private static String HEART_BEAT_MSG = "heartbeat";
    private static final Logger logger = LogManager.getLogger(MdcBroadcast.class);

    private static final String SOCKET_PREFIX = "MDC#";
    private static final AtomicInteger connectionIds = new AtomicInteger(0);
    private static final Set<MdcBroadcast> connections =
            new CopyOnWriteArraySet<>();

    private final String nickname;
    private Session session;

    public MdcBroadcast() {
        nickname = SOCKET_PREFIX + connectionIds.getAndIncrement();
    }


    @OnOpen
    public void start(Session session) {
        System.out.println("mdc start");
        this.session = session;
        connections.add(this);

        // MqttClientManager.getInstance().getMqttClientReceiver().locationReceive(this);
    }


    @OnClose
    public void end() {
        System.out.println("mdc end");
        connections.remove(this);
    }


    @OnMessage
    public void incoming(String message) {
        System.out.println("msg:" + message);

        if (HEART_BEAT_MSG.equals(message)) {
            try {
                session.getBasicRemote().sendText(HEART_BEAT_MSG);
            } catch (Exception e) {

                e.printStackTrace();
            }
        }
    }

    @OnMessage
    public void heartbeat(PongMessage pongMessage) {
        System.out.println("pong msg");

    }


    @OnError
    public void onError(Throwable t) throws Throwable {
        logger.error("Websocket Error: " + t.toString(), t);
        //出错后关闭连接
        processError(this);
    }

    public static void broadcast(String jsonStr) {


        System.out.println("mdc broadcast:" + jsonStr);

        List<MdcBroadcast> errorClientList = null;
        for (MdcBroadcast client : connections) {
            try {
                synchronized (client) {
                    if (!client.session.isOpen()) {
                        if (errorClientList == null) {
                            errorClientList = new ArrayList<>();
                            errorClientList.add(client);
                        }
                        continue;
                    }
                    client.session.getBasicRemote().sendText(jsonStr);
                }
            } catch (IOException e) {
                logger.debug("Websocket Error: Failed to send message to client", e);
                if (errorClientList == null) {
                    errorClientList = new ArrayList<>();
                    errorClientList.add(client);
                }
            }
        }

        if (errorClientList != null) {
            for (MdcBroadcast client : errorClientList) {
                processError(client);
            }
        }
    }


    static void broadcast(String msg, ArrayNode nodes) {

        String jsonStr = JsonMsg.asString(msg, true, nodes);

        broadcast(jsonStr);
    }

    static synchronized void processError(MdcBroadcast client) {
        connections.remove(client);
        try {
            client.session.close();
        } catch (IOException e1) {
            e1.printStackTrace();
        }
    }


    static synchronized void processError(List<MdcBroadcast> errorClientList) {

        if (errorClientList == null) {
            return;
        }

        for (MdcBroadcast client : errorClientList) {
            connections.remove(client);
            try {
                client.session.close();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }
    }
}